package com.owen.flink.batch.scala.relational

import com.owen.flink.batch.java.relational.util.WebLogData
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala.{ExecutionEnvironment, _}
import org.apache.flink.util.Collector

object WebLogAnalysis {

  def main(args: Array[String]): Unit = {
    val params: ParameterTool = ParameterTool.fromArgs(args)

    // set up execution environment
    val env = ExecutionEnvironment.getExecutionEnvironment

    // make parameters available in the web interface
    env.getConfig.setGlobalJobParameters(params)

    val document=getDocumentsDataSet(env,params)

    val ranks=getRanksDataSet(env,params)

    val visits=getVisitsDataSet(env,params)

    val documentFilter = document.filter(doc => doc._2.contains(" editors ") && doc._2.contains(" oscillations "))

    val ranksFilter = ranks.filter(rank => rank._1 > 40)

    val filteredVisits = visits
      .filter(visit => {
        val year=visit._2.substring(0, 4).toInt
        year == 2007 || year==2009 || year==2006 || year==2000
      })

    val joinDocsRanks = documentFilter.join(ranksFilter).where(0).equalTo(1) {
      (doc, rank) => rank
    }.withForwardedFieldsSecond("*")

    val result = joinDocsRanks.coGroup(filteredVisits).where(1).equalTo(0)  {
      (
        ranks: Iterator[(Int, String, Int)],
        visits: Iterator[(String, String)],
        out: Collector[(Int, String, Int)]) =>
        if (visits.isEmpty) for (rank <- ranks) out.collect(rank)
    }.withForwardedFieldsFirst("*")

    // emit result
    if (params.has("output")) {
      result.writeAsCsv(params.get("output"), "\n", "|")
      env.execute("Scala WebLogAnalysis Example")
    } else {
      println("Printing result to stdout. Use --output to specify output path.")
      result.print()
    }

  }

  private def getDocumentsDataSet(env: ExecutionEnvironment, params: ParameterTool)={
    println("Executing WebLogAnalysis example with default documents data set.")
    println("Use --documents to specify file input.")
    val documents = WebLogData.DOCUMENTS map {
      case Array(x, y) => (x.asInstanceOf[String], y.asInstanceOf[String])
    }
    env.fromCollection(documents)
  }

  private def getRanksDataSet(env: ExecutionEnvironment, params: ParameterTool) = {
    println("Executing WebLogAnalysis example with default ranks data set.")
    println("Use --ranks to specify file input.")
    val ranks = WebLogData.RANKS map {
      case Array(x, y, z) => (x.asInstanceOf[Int], y.asInstanceOf[String], z.asInstanceOf[Int])
    }
    env.fromCollection(ranks)
  }

  private def getVisitsDataSet(env: ExecutionEnvironment, params: ParameterTool) = {
    println("Executing WebLogAnalysis example with default visits data set.")
    println("Use --visits to specify file input.")
    val visits = WebLogData.VISITS map {
      case Array(x, y) => (x.asInstanceOf[String], y.asInstanceOf[String])
    }
    env.fromCollection(visits)
  }

}
